package net.wicp.tams.common.others.kafka;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;

import lombok.extern.slf4j.Slf4j;

/***
 * common.others.kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer
 * common.others.kafka.value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
 * 
 * @author Administrator
 *
 */
@Slf4j
public class KafkaAssitInst {
	private static KafkaAssitInst inst = new KafkaAssitInst();

	private Map<String, KafkaProducer<String, ?>> map = new HashMap<>();

	private KafkaAssitInst() {
	}

	public static KafkaAssitInst getInst() {
		return inst;
	}

	public <T> KafkaProducer<String, T> getKafkaProducer(Class<T> classz) {
		String key = classz.getName();
		if (map.isEmpty() || !map.containsKey(key)) {
			Properties prop = KafkaTools.getProps(true);
			String valueprop = KafkaTools.getValueProp(classz, true);
			prop.put("value.serializer", valueprop);
			prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
			log.info("kafka producer 参数:");
			for (Object propele : prop.keySet()) {
				log.info("{}:{}", propele, prop.get(propele));
			}
			// link和batch不用加
			KafkaProducer<String, T> kafkaProducer = new KafkaProducer<>(prop);
			map.put(key, kafkaProducer);
			return kafkaProducer;
		} else {
			return (KafkaProducer<String, T>) map.get(key);
		}
	}

	public void startkafkaCousumerS(String groupId, String topic, IConsumer<String> doConsumer, int hostNum) {
		KafkaConsumerGroup<String> group = new KafkaConsumerGroupS(groupId, topic, doConsumer, hostNum);
		group.start();
	}

	public void startkafkaCousumerS(String topic, IConsumer<String> doConsumer, int hostNum) {
		KafkaConsumerGroup<String> group = new KafkaConsumerGroupS(topic, doConsumer, hostNum);
		group.start();
	}

	public void startkafkaCousumerB(String groupId, String topic, IConsumer<byte[]> doConsumer, int hostNum) {
		KafkaConsumerGroup<byte[]> group = new KafkaConsumerGroupB(groupId, topic, doConsumer, hostNum);
		group.start();
	}

	public void startkafkaCousumerB(String topic, IConsumer<byte[]> doConsumer, int hostNum) {
		KafkaConsumerGroup<byte[]> group = new KafkaConsumerGroupB(topic, doConsumer, hostNum);
		group.start();
	}

	public void startKafkaConsumerThreadB(String groupId, String topic, IConsumer<byte[]> doConsumer, int threadNum) {
		KafkaConsumerThreadB group = new KafkaConsumerThreadB(groupId, topic, doConsumer);
		group.start(threadNum);
	}

	public void startKafkaConsumerThreadB(String topic, IConsumer<byte[]> doConsumer, int threadNum) {
		KafkaConsumerThreadB group = new KafkaConsumerThreadB(topic, doConsumer);
		group.start(threadNum);
	}

	public void startKafkaConsumerThreadS(String groupId, String topic, IConsumer<String> doConsumer, int threadNum) {
		KafkaConsumerThreadS group = new KafkaConsumerThreadS(groupId, topic, doConsumer);
		group.start(threadNum);
	}

	public void startKafkaConsumerThreadS(String topic, IConsumer<String> doConsumer, int threadNum) {
		KafkaConsumerThreadS group = new KafkaConsumerThreadS(topic, doConsumer);
		group.start(threadNum);
	}

}
